package com.distributed.master;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;

import static com.distributed.master.MasterServer.region_names;
import static com.distributed.master.MasterServer.regions_num;

public class RegionServerConnector extends Thread {
    private final ServerSocket server_socket;

    public RegionServerConnector(int port) throws IOException {
        server_socket = new ServerSocket(port);
        server_socket.setSoTimeout(Integer.MAX_VALUE);
    }

    /*
     * Author: Xinbei Wu
     * Function: connect with region server, read in region ip, output region name
     *  */
    @Override
    public void run() {
        while (true) {
            try {
                Socket socket = server_socket.accept();
                DataInputStream input_stream = new DataInputStream(socket.getInputStream());

                String response;
                String url = input_stream.readUTF();

                // ip : name
                if (region_names.containsKey(url)) { // region already exists
                    System.out.println("[INFO] " + url + " reconnected.");
                    response = region_names.get(url);
                } else { // new region
                    response = "/server-" + regions_num; // region name
                    // update information
                    regions_num++;
                    region_names.put(url, response);
                }

                // output
                DataOutputStream output_stream = new DataOutputStream(socket.getOutputStream());
                output_stream.writeUTF(response);

                // close
                socket.close();
            } catch (SocketTimeoutException s) {
                System.out.println("[ERROR] Socket times out.");
                break;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
